/* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */
/* Copyright 2002-2003 by Mark Wolgemuth, All rights reserved.             */
/* Copyright 2002-2003 by Employease, Inc.  All rights reserved.           */
/*                                                                         */
/* Redistribution and use in source and binary forms, with or without      */
/* modification, are permitted provided that the following conditions      */
/* are met:                                                                */
/*                                                                         */
/*    1. Redistributions of source code must retain the above copyright    */
/* notice, this list of conditions and the following disclaimer.           */
/*    2. Redistributions in binary form must reproduce the above copyright */
/* notice, this list of conditions and the following disclaimer in the     */
/* documentation and/or other materials provided with the distribution.    */
/*                                                                         */
/* THIS SOFTWARE IS PROVIDED BY THE MOD_ATHENA PROJECT ``AS IS'' AND ANY   */
/* EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE       */
/* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR      */
/* PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE FREEBSD PROJECT OR        */
/* CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,   */
/* EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,     */
/* PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR      */
/* PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY     */
/* OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT            */
/* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE   */
/* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.    */
/* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */

#include "athd.h"

/* begin_privates */
static athd_request * _athd_req_alloc(apr_pool_t *pool);
static char * _athd_req_gen_update_phys(apr_pool_t *p, 
					const ath_upd_phys *phys);
static char * _athd_req_gen_update_farm(apr_pool_t *p, 
					const ath_upd_farm *farm);
static char * _athd_req_gen_serial_phys(apr_pool_t *p, const char *serial);
static char * _athd_req_gen_serial_farm(apr_pool_t *p, const char *serial);
static char * _athd_req_gen_mirror_phys(apr_pool_t *p, const char *serial);
static char * _athd_req_gen_mirror_farm(apr_pool_t *p, const char *serial);
static apr_status_t _athd_req_response_ready(athd_request *req);
/* end_privates */

/* #define ATHD_SHOW_PHYS_FMT "host:%h|time:%t|on:%o|up:%u|in:%i|cpu:%c|net:%n|" \ */
/*   "ld:%l|mem:%m|dsk:%d|cus0:%0|cus1:%1|cus2:%2|cus3:%3|cus4:%4|cus5:%5|" \ */
/*   "cus6:%6|cus7:%7|cus8:%8|cus9:%9|farms:%f" */
/* #define ATHD_SHOW_FARM_FMT "name:%n|on:%o|algo:%a|hitadd:%h|expect:%e|" \ */
/*   "ttl:%t|rttl:%r|chosen:%c|downurl:%d|offlineurl:%D|members:%m" */

/* #define ATHD_LIST_PHYS_FMT "%h|%t|%o|%u|%i" */
/* #define ATHD_LIST_FARM_FMT "%n|%o|%m" */

athd_request *
athd_req_serf_req_build(athd_request *req)
{
    serf_filter_t *filter;	  
    ASSERT(req);
    req->serf_request = serf_create_request(req->pool);
    req->serf_request->ctx = req;
    req->serf_request->source = athd_req_handler_in;
    req->serf_request->handler = athd_req_handler_out;
    req->serf_request->method = ATHD_HTTP_METHOD;
    req->serf_request->uri = req->uri;
    /* request filters */
    filter = serf_add_filter(req->serf_request->request_filters, 
			     "USER_AGENT", req->pool);
    filter = serf_add_filter(req->serf_request->request_filters, 
			     "HOST_HEADER", req->pool);
    filter->ctx = req->serf_request;
    filter = serf_add_filter(req->serf_request->request_filters, 
			     "HTTP_HEADERS_OUT", req->pool);
    /* response filters */
    filter = serf_add_filter(req->serf_request->response_filters, 
			     "HTTP_STATUS_IN", req->pool);
    filter = serf_add_filter(req->serf_request->response_filters, 
			     "HTTP_HEADERS_IN", req->pool);
    filter = serf_add_filter(req->serf_request->response_filters, 
			     "HTTP_DECHUNK", req->pool);
    filter = serf_add_filter(req->serf_request->response_filters,
			     "HTTP_RESP_LEN", req->pool);
    filter->ctx = req;
/* 	  filter = serf_add_filter(req->serf_request->response_filters,  */
/* 				   "DEBUG_RESPONSE", req->pool); */
    return req;
}

athd_request *
athd_req_create(apr_pool_t *pool, athd_engine *engine, 
		athd_req_type type, const void *data)
{
    int rv;
    athd_request *req;
    ASSERT(pool);
    ASSERT(engine);
    req = _athd_req_alloc(pool);
    req->type = type;
    req->engine = engine;
    req->content_length = 0;

#define MATCH(t) (type == t)

    if MATCH(ATHD_REQ_UPDATE_PHYS)
      req->url = _athd_req_gen_update_phys(req->pool, 
					   (const ath_upd_phys *)data); 
    else if MATCH(ATHD_REQ_UPDATE_FARM)
      req->url = _athd_req_gen_update_farm(req->pool, 
					   (const ath_upd_farm *)data);
    else if MATCH(ATHD_REQ_SERIAL_PHYS)
    {
	req->url = _athd_req_gen_serial_phys(req->pool, (const char *)data);
	req->cacheable = 1;
    }
    else if MATCH(ATHD_REQ_SERIAL_FARM)
    {
	req->url = _athd_req_gen_serial_farm(req->pool, (const char *)data);
	req->cacheable = 1;
    }
    else if MATCH(ATHD_REQ_MIRROR_PHYS)
      req->url = _athd_req_gen_mirror_phys(req->pool, (const char *)data);
    else if MATCH(ATHD_REQ_MIRROR_FARM)
      req->url = _athd_req_gen_mirror_farm(req->pool, (const char *)data);
    else 
	req->url = ATH_REQ_HELLO;

    req->url = apr_pstrcat(req->pool, engine->url, req->url, NULL);
    rv = apr_uri_parse(req->pool, req->url, req->uri);
    if (!rv)
      {
	  if (!req->uri->port)
	      req->uri->port = 
		apr_uri_default_port_for_scheme(req->uri->scheme);
	  /* not sending query string in gets? blah, rebuild it */
	  /* is this a bug in apr_uri_* or serf? */
	  if (req->uri->query)
	      req->uri->path = apr_pstrcat(req->pool, req->uri->path, 
					   "?", req->uri->query, NULL);
	  /* athd_req_serf_req_build(req); always do this in run req */
      }
    else
      {
	  athd_log_err(ATHD_WARN, rv, "Request: uri error: ");
	  athd_req_destroy(&req);
      }
    return req;
}

athd_request *
athd_req_create_for_job(athd_job *job, athd_engine *eng,
			athd_req_type type, const void *data)
{
    int rv;
    athd_request *req;
    athd_engine *engine;
    ASSERT(job);
    ASSERT(job->pool);

    if (eng)
	engine = eng;
    else
      {
	  ASSERT(job->engine);
	  engine = job->engine;
      }
    req = athd_req_create(job->pool, engine, type, data);
    req->job = job;
    return req;
}

void
athd_req_destroy(athd_request **req)
{
    apr_pool_t *pooltmp;
    ASSERT((*req));
    /* alert any waiting threads to destruction */
    /* this could be catastrophic */
    pthread_cond_broadcast((*req)->response_cv);
    pthread_mutex_unlock((*req)->response_mx);
    pooltmp = (*req)->pool;
    apr_pool_destroy(pooltmp);
    (*req) = NULL;
}

/* asynchronous request setup callback */
apr_status_t
athd_req_handler_in(apr_bucket_brigade *brigade, serf_request_t *request,
			apr_pool_t *pool)
{
    apr_bucket *bucket;
    athd_request *athreq;
    ASSERT(pool);
    ASSERT(request);
    athreq = request->ctx;
    ASSERT(athreq);
    ASSERT(athreq->conn);

    /* now we're REALLY going to use it, we're in serf section */
    athd_connection_acquire(athreq->conn);

    bucket = serf_bucket_request_line_create(request->method,
					     request->uri->path,
					     ATHD_HTTP_VERS,
					     pool,
					     brigade->bucket_alloc);
    APR_BRIGADE_INSERT_HEAD(brigade, bucket);
    
    if (!athreq->engine->keep_alive)
      {
	  bucket = serf_bucket_header_create("Connection",
					     "Close",
					     pool, brigade->bucket_alloc);
	  APR_BRIGADE_INSERT_TAIL(brigade, bucket);
      }

    return APR_SUCCESS;
}



/* asynchronous request response callback */
apr_status_t
athd_req_handler_out(serf_response_t *response, apr_pool_t *pool)
{
    apr_status_t rv = APR_SUCCESS;
    /* get orig request */
    serf_request_t *request = response->request; 
    /* get our athd request */
    athd_request *athreq = NULL;
    apr_bucket *bucket;
    apr_size_t readsofar = 0;
    char *str_response = NULL;
    ASSERT(response);
    ASSERT(request);
    ASSERT(pool);
    athreq = (athd_request *) request->ctx;
    ASSERT(athreq);

    /* Print anything that isn't metadata. */


    for (bucket = APR_BRIGADE_FIRST(response->entity);
	 ((!rv)
	  && (bucket != APR_BRIGADE_SENTINEL(response->entity))
	  && (readsofar < athreq->content_length)
	  );
         bucket = APR_BUCKET_NEXT(bucket)) 
      {
	  if (!APR_BUCKET_IS_METADATA(bucket)) 
	    {
		apr_size_t read;
		const char *buf;
		rv = apr_bucket_read(bucket, &buf, &read, APR_BLOCK_READ);
		if (!rv) 
		  {
		      if (str_response)
			{
			    char *str_continue = 
			      apr_pcalloc(athreq->pool, 
					  sizeof(char)*(readsofar+read+1));
			    apr_cpystrn(str_continue, str_response, 
					readsofar + 1);
			    memcpy(str_continue + readsofar, buf, read);
			    str_response = str_continue;
			}
		      else
			{
			    str_response = 
			      apr_pcalloc(athreq->pool, sizeof(char)*(read+1));
			    memcpy(str_response, buf, read);
			    *(str_response + read + 1) = (int)NULL;
			}
		      readsofar += read;
		      *(str_response + readsofar + 1) = (int)NULL;
		  }
		else
		    athd_log_err(ATHD_DEBUG, rv, "Read error: ");

		athd_log(ATHD_DEVEL, "Request: [%s] RECV: [%ld/%ld]", 
			 athreq->url,  readsofar, athreq->content_length);
	    }
      }
    
    athreq->response_txt = str_response;

    /* signal any threads waiting other than job to go ahead with resp */
    _athd_req_response_ready(athreq);

    /* moved to _athd_engine_run_reqs */
    /* athd_job_push_response(athreq); */
    if (athreq->conn->serf_conn->socket)
	apr_brigade_cleanup(response->entity);
    /* connection post-process, serf is done with it */
    athd_connection_release(&athreq->conn);
    return rv;
}

apr_status_t
athd_req_response_wait(athd_request *req)
{
    return pthread_cond_wait(req->response_cv, req->response_mx);
}


/* privates */

static athd_request * 
_athd_req_alloc(apr_pool_t *pool)
{
    athd_request *req;
    apr_pool_t *rpool;
    ASSERT(pool);
    apr_pool_create(&rpool, pool);
    req = apr_pcalloc(rpool, sizeof(athd_request));
    req->pool = rpool;
    req->serf_request = NULL;
    req->uri = apr_pcalloc(req->pool, sizeof(apr_uri_t));
    req->engine = NULL;
    req->job = NULL;
    req->response_txt = NULL;
    req->tries = 0;
    req->expires = 0;
    req->conn = NULL;
    req->response_cv = apr_pcalloc(req->pool, sizeof(pthread_cond_t));
    apr_pool_cleanup_register(req->pool, req->response_cv,
			      pthread_cond_cleanup, apr_pool_cleanup_null);
    pthread_cond_init(req->response_cv, NULL);
    req->response_mx = apr_pcalloc(req->pool, sizeof(pthread_mutex_t));
    apr_pool_cleanup_register(req->pool, req->response_mx,
			      pthread_mutex_cleanup, apr_pool_cleanup_null);
    pthread_mutex_init(req->response_mx, NULL);
    /* the response mutex is locked to start */
    /* this gets cleared by implementer if no response is needed */
    /* or just ignored if athd_req_response_wait is never called */
    pthread_mutex_lock(req->response_mx);
    req->cacheable = 0;
    req->cached = 1;
    return req;
}

/* only supports updating singel phys->physs host */
static char * 
_athd_req_gen_update_phys(apr_pool_t *pool, const ath_upd_phys *phys)
{
    char *gen;
    char *host;
    int i;
    for (i = 0; i < ath_list_size(phys->physs); i++)
      {
	  ath_node *listctx;
	  ath_list_rewind_r(phys->physs, &listctx);
	  host = (char *) ath_list_getnext_r(phys->physs, &listctx);
	  if (!gen) /* need ATH_REQ_JOIN here in case other type req precedes */
	      gen = apr_pstrcat(pool, 
				ATH_REQ_JOIN, ATH_REQ_UPDATE, ATH_REQ_PHYS, 
				ATH_REQ_ARG_JOIN,
				apr_psprintf(pool, ",%c=", ATH_CASE_P_HOST),
				host, NULL);
	  else
	      gen = apr_pstrcat(pool, gen, 
				ATH_REQ_JOIN, ATH_REQ_UPDATE, ATH_REQ_PHYS, 
				ATH_REQ_ARG_JOIN, 
				apr_psprintf(pool, ",%c=", ATH_CASE_P_HOST),
				host, NULL);
	  if (phys->on_u)
	      gen = apr_pstrcat(pool, gen, 
				apr_psprintf(pool, ",%c=%d", 
					     ATH_CASE_ON, phys->on), NULL);
	  else if (phys->up_u)
	      gen = apr_pstrcat(pool, gen, 
				apr_psprintf(pool, ",%c=%d", 
					     ATH_CASE_UP, phys->up), NULL);
	  else if (ath_phys_upd_field_updated(phys, ATH_ALGO_CPU))
	      gen = apr_pstrcat(pool, gen, 
				apr_psprintf(pool, ",%c=%f", 
					     ATH_ALGO_CPU, phys->cpu), NULL);
	  else if (ath_phys_upd_field_updated(phys, ATH_ALGO_NET))
	      gen = apr_pstrcat(pool, gen, 
				apr_psprintf(pool, ",%c=%f",
					     ATH_ALGO_NET, phys->net), NULL);
	  else if (ath_phys_upd_field_updated(phys,ATH_ALGO_LOD))
	      gen = apr_pstrcat(pool, gen, 
				apr_psprintf(pool, ",%c=%f", 
					     ATH_ALGO_LOD, phys->ld), NULL);
	  else if (ath_phys_upd_field_updated(phys,ATH_ALGO_MEM))
	      gen = apr_pstrcat(pool, gen, 
				apr_psprintf(pool, ",%c=%f", 
					     ATH_ALGO_MEM, phys->mem), NULL);
	  else if (ath_phys_upd_field_updated(phys,ATH_ALGO_CU1))
	      gen = apr_pstrcat(pool, gen, 
				apr_psprintf(pool, ",%c=%f", 
					     ATH_ALGO_CU1, phys->cus1), NULL);
	  else if (ath_phys_upd_field_updated(phys,ATH_ALGO_CU2))
	      gen = apr_pstrcat(pool, gen, 
				apr_psprintf(pool, ",%c=%f", 
					     ATH_ALGO_CU2, phys->cus2), NULL);
	  else if (ath_phys_upd_field_updated(phys,ATH_ALGO_CU3))
	      gen = apr_pstrcat(pool, gen, 
				apr_psprintf(pool, ",%c=%f",
					     ATH_ALGO_CU3,phys->cus3), NULL);
      }
    return gen;
}

static char * 
_athd_req_gen_update_farm(apr_pool_t *pool, const ath_upd_farm *farm)
{
    char *gen;
    char *name;
    int i,j;
    for (i = 0; i < ath_list_size(farm->farms); i++)
      {
	  name = (char *) ath_list_getnext(farm->farms);
	  if (!gen) /* need ATH_REQ_JOIN here in case other type req precedes */
	      gen = apr_pstrcat(pool, 
				ATH_REQ_JOIN, ATH_REQ_UPDATE, ATH_REQ_FARM, 
				ATH_REQ_ARG_JOIN, 
				apr_psprintf(pool, ",%c=", ATH_CASE_F_NAME),
				name, NULL);
	  else
	      gen = apr_pstrcat(pool, gen, 
				ATH_REQ_JOIN, ATH_REQ_UPDATE, ATH_REQ_FARM, 
				ATH_REQ_ARG_JOIN, 
				apr_psprintf(pool, ",%c=", ATH_CASE_F_NAME),
				name, NULL);    
	  if (farm->on_u)
	      gen = apr_pstrcat(pool, gen, 
				apr_psprintf(pool, ",%c=%d", 
					     ATH_CASE_ON, farm->on), NULL);
	  if (farm->str_algo)
	      gen = apr_pstrcat(pool, gen, 
				apr_psprintf(pool, ",%c=", ATH_CASE_ALGO),
				farm->str_algo, NULL);
	  if (farm->hit_adds_to != (int) ATH_ALGO_NON)
	      gen = apr_pstrcat(pool, gen, 
				apr_psprintf(pool, ",%c=%s-%f", 
					     ATH_CASE_HITADDS,
					     ath_algo_type_to_str(farm->hit_adds_to), 
					     farm->hit_adds),
				NULL);
	  if (farm->expect_u)
	      gen = apr_pstrcat(pool, gen, 
				apr_psprintf(pool, ",%c=%d", 
					     ATH_CASE_EXPECT, farm->expect), 
				NULL);
	  if (farm->str_down_url)
	      gen = apr_pstrcat(pool, gen, apr_psprintf(pool, ",%c=", 
							ATH_CASE_DURL),
				farm->str_down_url,
				NULL);
	  if (farm->str_offline_url)
	      gen = apr_pstrcat(pool, gen, apr_psprintf(pool, ",%c=", 
							ATH_CASE_OURL),
				farm->str_offline_url,
				NULL);
	  for (j = 0; j < ath_list_size(farm->member_add); j++)
	    {
		char *member = (char *) ath_list_getnext(farm->member_add);
		gen = apr_pstrcat(pool, gen, 
				  apr_psprintf(pool, ",%c=", ATH_CASE_ADD_P), 
				  member, NULL);  
	    }
	  for (j = 0; j < ath_list_size(farm->member_rem); j++)
	    {
		char *member = (char *) ath_list_getnext(farm->member_rem);
		gen = apr_pstrcat(pool, gen, 
				  apr_psprintf(pool, ",%c=", ATH_CASE_REM_P),
				  member, NULL); 
	    }


      }
    return gen;
}

static char * 
_athd_req_gen_serial_phys(apr_pool_t *pool, const char *phys)
{
    char *gen;
    gen = apr_pstrcat(pool, ATH_REQ_STATUS, ATH_REQ_PHYS, ATH_REQ_ARG_JOIN,
		      apr_psprintf(pool, "%c=%s,%c=%%%c", 
				   ATH_CASE_P_HOST, phys, 
				   ATH_CASE_FMT, ATH_CASE_SERIAL),
		      NULL);
    return gen;
}

static char * 
_athd_req_gen_serial_farm(apr_pool_t *pool, const char *farm)
{
    char *gen;
    gen = apr_pstrcat(pool, ATH_REQ_STATUS, ATH_REQ_FARM, ATH_REQ_ARG_JOIN,
		      apr_psprintf(pool, "%c=%s,%c=%%%c", 
				   ATH_CASE_F_NAME, farm, 
				   ATH_CASE_FMT, ATH_CASE_SERIAL),
		      NULL);
    return gen;
}

static char *
_athd_req_gen_mirror_phys(apr_pool_t *pool, const char *phys_serial)
{
    char *path = NULL;
    path = apr_pstrcat(pool, ATH_REQ_MIRROR, ATH_REQ_PHYS, ATH_REQ_ARG_JOIN, 
		       phys_serial, NULL);
    return path;
}

static char *
_athd_req_gen_mirror_farm(apr_pool_t *pool, const char *farm_serial)
{
    char *path = NULL;
    path = apr_pstrcat(pool, ATH_REQ_MIRROR, ATH_REQ_FARM, ATH_REQ_ARG_JOIN, 
		       farm_serial, NULL);
    return path;
}

static apr_status_t
_athd_req_response_ready(athd_request *req)
{
    //pthread_mutex_lock(req->response_mx);
    pthread_cond_broadcast(req->response_cv);
    return pthread_mutex_unlock(req->response_mx);
}
